This notebook requires Spark 2.2+ to be pip installed your env.
This workshop takes you through building a simple decision tree model using Spark's ML pipeline interface.
The different algorithms options are contained in http://spark.apache.org/docs/latest/api/python/pyspark.ml.html The Spark ML documentation is at http://spark.apache.org/docs/latest/ml-pipeline.html
If you're looking for Spark books why not consider Learning Spark or High Performance Spark (co-authored by the author of this notebook) -- http://amzn.to/2etDd0L
Need more than the algorithms available in Spark? Or just have some time ot kill? Why not join us in https://github.com/sparklingpandas/sparklingml and add your own algorithm! :)
In [12]:
from pyspark import SparkContext
from pyspark.conf import SparkConf
from pyspark.sql.session import SparkSession
In [14]:
# Configure your SparkContext. You can set the master, by default this will use a local master unless you
# add setMaster or change your enviorment variables.
conf = SparkConf().setAppName("intro-to-ml")
# We use get or create here so that if the cell is evaluated multiple times we don't get multiple SparkContexts.
sc = SparkContext.getOrCreate(conf)
In [15]:
sqlContext = SparkSession.builder.getOrCreate()
Now we start by downloading loading some data which is in csv format so its a good thing we got that csv package included already for us.
Note: the data is a modified version of https://archive.ics.uci.edu/ml/datasets/Adult
In [18]:
# If your running this on a cluster you will need to copy the data file into HDFS or whatever cluster file system
# you are using
df = sqlContext.read.format("csv").option("header", "true").load("resources/adult.data")
In [19]:
df.cache()
Out[19]:
In [20]:
df.head()
Out[20]:
So as we can see Spark has simply loaded all of the values as strings since we haven't specified another schema. We can isntead ask it to infer the schema and also handle this extra space magic:
In [21]:
df = sqlContext.read.format("csv").option("header", "true").option("inferSchema", "true").load("resources/adult.data")
In [22]:
df.head()
Out[22]:
In [23]:
df.cache()
Out[23]:
Spark has two different machine learning libraries. We will use the new Dataframe based one which is defined in pyspark.ml. Let's import the basics as well as DecisionTreeClassifier :)
In [34]:
from pyspark.mllib.linalg import Vectors
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.param import Param, Params
from pyspark.ml.feature import Bucketizer, VectorAssembler, StringIndexer
from pyspark.ml import Pipeline
The first step is prepairing the features, here we are just choosing existing numeric features:
In [35]:
assembler = VectorAssembler(inputCols=["age", "education-num"], outputCol="feautres")
Now the pipeline only works on doubles, so we need to take our category and turn it into a double. The StringIndexer will do this for us:
In [36]:
indexer = StringIndexer(inputCol="category").setOutputCol("category-index")
In [37]:
pipeline = Pipeline().setStages([assembler, indexer])
We then need to "fit" our pipeline. This allows the StringIndexer to determine what strings will be assigned what indexes in the eventual transformation:
In [38]:
model=pipeline.fit(df)
We then transform our data into the prepaired format for our machine learning model to work on:
In [39]:
prepared = model.transform(df)
In [40]:
prepared.head()
Out[40]:
In [41]:
dt = DecisionTreeClassifier(labelCol = "category-index", featuresCol="feautres")
And now we fit on the prepared data
In [42]:
dt_model = dt.fit(prepared)
In [43]:
dt_model
Out[43]:
But manually chaining this steps together makes it difficult to experiment. Instead we can put it together in a pipeline:
In [44]:
pipeline_and_model = Pipeline().setStages([assembler, indexer, dt])
pipeline_model = pipeline_and_model.fit(df)
In [45]:
dt_model.transform(prepared).select("prediction", "category-index").take(20)
Out[45]:
In [46]:
pipeline_model.transform(df).select("prediction", "category-index").take(20)
Out[46]:
While this shows us the results, it isn't super easy to read. Thankfully we can inverse these.
In [47]:
labels = list(pipeline_model.stages[1].labels)
In [48]:
from pyspark.ml.feature import IndexToString
inverter = IndexToString(inputCol="prediction", outputCol="prediction-label", labels=labels)
In [49]:
inverter.transform(pipeline_model.transform(df)).select("prediction-label", "category").take(20)
Out[49]:
In [50]:
pipeline_model.stages[2]
Out[50]:
In [51]:
from pyspark.sql.functions import *
df.groupBy("age").agg(min("hours-per-week"), avg("hours-per-week"), max("capital-gain"))
Out[51]:
In [52]:
from pyspark.sql.window import Window
windowSpec = Window.partitionBy("age").orderBy("capital-gain").rowsBetween(-100, 100)
In [53]:
df.select(df["age"], df['capital-gain'], avg("capital-gain").over(windowSpec)).orderBy(desc("capital-gain")).show()
Now it's over to you to take it to the next steps!
First:
After you've figured out how to measure effectiveness, it's time to see what we can do to improve:
In [ ]: